package com.ververica.cdc.guass.source.hybrid;

import com.ververica.cdc.guass.source.jdbc.SnapShotSplit;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataOutputSerializer;

import java.io.IOException;
import java.sql.Date;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;

/**
 * HybridSourceSplitSerializer 用于序列化和反序列化 HybridSourceSplit 对象。
 */


/**
 * HybridSourceSplitSerializer 用于序列化和反序列化 HybridSourceSplit 对象。
 */
public class HybridSourceSplitSerializer implements SimpleVersionedSerializer<HybridSourceSplit> {

    // Type identifiers
    private static final int TYPE_DOUBLE = 1;
    private static final int TYPE_STRING = 2;
    private static final int TYPE_DATE = 3;
    private static final int TYPE_TIMESTAMP = 4;

    @Override
    public int getVersion() {
        return 1;
    }

    @Override
    public byte[] serialize(HybridSourceSplit split) throws IOException {
        DataOutputSerializer out = new DataOutputSerializer(256);
        out.writeUTF(split.splitId());
        out.writeBoolean(split.isSnapshot());
        out.writeInt(split.getSourceIndex()); // 序列化 sourceIndex
        if (split.isSnapshot()) {
            SnapShotSplit snapShotSplit = (SnapShotSplit) split.getSplitDetail();

            // 序列化 tableName
            out.writeUTF(snapShotSplit.getTableName());

            // 序列化 splitColumns
            List<String> splitColumns = snapShotSplit.getSplitColumns();
            out.writeInt(splitColumns.size());
            for (String column : splitColumns) {
                out.writeUTF(column);
            }

            // 序列化 splitStart
            Object[] splitStarts = snapShotSplit.getSplitStart();
            out.writeInt(splitStarts.length);
            for (Object start : splitStarts) {
                serializeValue(out, start);
            }

            // 序列化 splitEnd
            Object[] splitEnds = snapShotSplit.getSplitEnd();
            out.writeInt(splitEnds.length);
            for (Object end : splitEnds) {
                serializeValue(out, end);
            }

            // 序列化 snapshotId 和 snapshotTimestamp
            out.writeUTF(snapShotSplit.getSnapshotId());
            out.writeLong(snapShotSplit.getSnapshotTimestamp());
        }
        return out.getCopyOfBuffer();
    }

    @Override
    public HybridSourceSplit deserialize(int version, byte[] serialized) throws IOException {
        if (version != 1) {
            throw new IOException("Unsupported version: " + version);
        }
        DataInputDeserializer in = new DataInputDeserializer(serialized);
        String splitId = in.readUTF();
        boolean isSnapshot = in.readBoolean();
        int sourceIndex = in.readInt(); // 反序列化 sourceIndex
        Object splitDetail = null;
        if (isSnapshot) {
            String tableName = in.readUTF();

            // 反序列化 splitColumns
            int columnsSize = in.readInt();
            List<String> splitColumns = new ArrayList<>();
            for (int i = 0; i < columnsSize; i++) {
                splitColumns.add(in.readUTF());
            }

            // 反序列化 splitStart
            int startSize = in.readInt();
            Object[] splitStart = new Object[startSize];
            for (int i = 0; i < startSize; i++) {
                splitStart[i] = deserializeValue(in);
            }

            // 反序列化 splitEnd
            int endSize = in.readInt();
            Object[] splitEnd = new Object[endSize];
            for (int i = 0; i < endSize; i++) {
                splitEnd[i] = deserializeValue(in);
            }

            // 反序列化 snapshotId 和 snapshotTimestamp
            String snapshotId = in.readUTF();
            long snapshotTimestamp = in.readLong();

            splitDetail = new SnapShotSplit(tableName, splitColumns, splitStart, splitEnd, snapshotId, snapshotTimestamp);
        }
        return new HybridSourceSplit(splitId, isSnapshot, splitDetail, sourceIndex);
    }

    /**
     * Helper method to serialize different value types using type identifiers.
     */
    private void serializeValue(DataOutputSerializer out, Object value) throws IOException {
        if (value instanceof Double) {
            out.writeInt(TYPE_DOUBLE);  // Type identifier for Double
            out.writeDouble((Double) value);
        } else if (value instanceof String) {
            out.writeInt(TYPE_STRING); // Type identifier for String
            out.writeUTF((String) value);
        } else if (value instanceof Date) {
            out.writeInt(TYPE_DATE);   // Type identifier for Date
            out.writeLong(((Date) value).getTime());
        } else if (value instanceof Timestamp) {
            out.writeInt(TYPE_TIMESTAMP); // Type identifier for Timestamp
            out.writeLong(((Timestamp) value).getTime());
        } else {
            throw new IOException("Unsupported value type: " + value.getClass().getName());
        }
    }

    /**
     * Helper method to deserialize different value types based on the type identifier.
     */
    private Object deserializeValue(DataInputDeserializer in) throws IOException {
        int typeId = in.readInt();
        switch (typeId) {
            case TYPE_DOUBLE:
                return in.readDouble();
            case TYPE_STRING:
                return in.readUTF();
            case TYPE_DATE:
                return new Date(in.readLong());
            case TYPE_TIMESTAMP:
                return new Timestamp(in.readLong());
            default:
                throw new IOException("Unsupported value type with identifier: " + typeId);
        }
    }
}
